package com.stm.bi.repository;


import com.alibaba.fastjson.JSON;
import com.stm.bi.core.util.DirPathUtil;
import com.stm.bi.jdbc.DataSourceEnum;
import com.stm.bi.jdbc.DorisJdbcUtils;
import com.stm.bi.repository.dataobject.DorisLoadRecordDO;
import com.stm.framework.context.TraceContext;
import com.stm.framework.core.util.*;
import com.stm.framework.dbcp.DataSourceConfig;
import jakarta.annotation.Resource;
import org.apache.commons.codec.binary.Base64;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPut;
import org.apache.http.entity.FileEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.DefaultRedirectStrategy;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.http.HttpHeaders;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;

import javax.sql.DataSource;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;


@Component
public class VerToBetaDorisLoadRepository {
    private static final Logger log = LoggerFactory.getLogger(VerToBetaDorisLoadRepository.class);
    @Value("${beta-doris.url:}")
    private String dorisUrl;
    @Value("${beta-doris.username:}")
    private String username;
    @Value("${beta-doris.passowrd:}")
    private String password;
    @Value("${spring.profiles.active}")
    private String env;

    @Resource
    private DorisLoadRecordRepository dorisLoadRecordRepository;


    private final static HttpClientBuilder httpClientBuilder = HttpClients
            .custom()
            .setRedirectStrategy(new DefaultRedirectStrategy() {
                @Override
                protected boolean isRedirectable(String method) {
                    // 如果连接目标是 FE，则需要处理 307 redirect。
                    return true;
                }
            });


    public void putLoad(String schema, String tableName,File file,String columns,String separator) {
        if(!"ver".equals(env)){
            return;
        }
        try (CloseableHttpClient client = httpClientBuilder.build()) {
            HttpPut put = new HttpPut(dorisUrl+"/api/"+schema+"/"+tableName+"/_stream_load");
            put.setHeader(HttpHeaders.EXPECT, "100-continue");
            put.setHeader(HttpHeaders.AUTHORIZATION, basicAuthHeader(username, password));
            put.setHeader("Expect","100-continue");
            put.setHeader("column_separator",separator);
            put.setHeader("columns",columns);
            // 设置错误行过滤比例
            put.setHeader("max_filter_ratio","0.1");
            // 设置导入文件。
            // 这里也可以使用 StringEntity 来传输任意数据。
            FileEntity entity = new FileEntity(file);
            put.setEntity(entity);


            try (CloseableHttpResponse response = client.execute(put)) {
                String loadResult = "";
                if (response.getEntity() != null) {
                    loadResult = EntityUtils.toString(response.getEntity());
                }
                final int statusCode = response.getStatusLine().getStatusCode();
                if (statusCode != 200) {
                    throw new IOException(String.format("导入失败. status: %s load result: %s", statusCode, loadResult));
                }
                DorisLoadRecordDO dorisLoadRecord = JSON.parseObject(loadResult, DorisLoadRecordDO.class);
                dorisLoadRecord.setLoadTime(LocalDateTime.now());
                dorisLoadRecord.setTraceId(TraceContext.getId());
                dorisLoadRecord.setTableName(tableName);
                dorisLoadRecord.setSchemaName(schema);
                dorisLoadRecordRepository.create(dorisLoadRecord);
                log.info("Get load result: {}", loadResult);
                if(dorisLoadRecord.getNumberFilteredRows() >0){
                    dorisLoadRecord.setLoadTime(LocalDateTime.now());
                    dorisLoadRecord.setTraceId(TraceContext.getId());
                    dorisLoadRecord.setTableName(tableName);
                    dorisLoadRecord.setSchemaName(schema);
                    dorisLoadRecord.setStatus("Fail");
                    dorisLoadRecordRepository.create(dorisLoadRecord);
                    throw new RuntimeException(dorisLoadRecord.getMessage());
                }
            }
        }catch (Exception e) {
            log.error("上传文件：{}到Doris报错{}",file.getName(),e);
            DorisLoadRecordDO dorisLoadRecord = new DorisLoadRecordDO();
            dorisLoadRecord.setLoadTime(LocalDateTime.now());
            dorisLoadRecord.setTraceId(TraceContext.getId());
            dorisLoadRecord.setTableName(tableName);
            dorisLoadRecord.setSchemaName(schema);
            dorisLoadRecord.setStatus("Fail");
            dorisLoadRecord.setMessage(e.getMessage());
            dorisLoadRecordRepository.create(dorisLoadRecord);
            throw new RuntimeException(e);
        }
    }

    private String basicAuthHeader(String username, String password) {
        final String tobeEncode = username + ":" + password;
        //log.info("用户名：密码{}",tobeEncode);
        byte[] encoded = Base64.encodeBase64(tobeEncode.getBytes(StandardCharsets.UTF_8));
        return "Basic " + new String(encoded);
    }

}